package com.david;

import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class ReactorTest {
    public static void main(String[] args) {

        String[] s=new String[]{"xx","oo"};
        //just 已知元素数量和内容
        //flux1 发布者（被观察者）
        Flux<String> flux1 = Flux.just(s);
        //println 订阅者（观察者）
        flux1.subscribe(System.out::println);


        Flux<String> flux2 = Flux.just("xx", "xxx");
        flux2.subscribe(System.out::println);


        List<String> list= Arrays.asList("hello","world");
        Flux<String> flux3 = Flux.fromIterable(list);
        flux3.subscribe(System.out::println);


        Stream<String> stream=Stream.of("hi","hello");
        Flux<String> flux4 = Flux.fromStream(stream);
        flux4.subscribe(System.out::println);


        Flux<Integer> range = Flux.range(0, 5);
        range.subscribe(System.out::println);


        //合并
        Flux<String> mergeWith = flux3.mergeWith(flux4);
        mergeWith.subscribe(System.out::println);

        //结合为元祖
        Flux<String> source1 = Flux.just("111", "world", "333");
        Flux<String> source2 = Flux.just("2221","xxx");
        Flux<Tuple2<String,String>> zip=source1.zipWith(source2);
        zip.subscribe(tuple->{
            System.out.println(tuple.getT1()+"->"+tuple.getT2());
        });

        //跳过两个
        Flux<String> flux = Flux.just("111", "222", "333");
        Flux<String> skip = flux.skip(2);
        skip.subscribe(System.out::println);

        //拿前两个
        Flux<String> flux7  = Flux.just("111", "222", "333");
        Flux<String> skip2 = flux7.take(2);
        skip2.subscribe(System.out::println);


        //同步动态创建 next只能被调用一次
        Flux.generate(sink->{
            sink.next("xx");
            sink.complete();
        }).subscribe(System.out::println);

        //异步动态创建
        Flux.create(sink->{
            for (int i=0;i<10;i++){
                sink.next("xxoo"+i);
            }
            sink.complete();
        }).subscribe(System.out::println);

    }
}
